package application

import (
	"context"
	"encoding/csv"
	"encoding/json"
	"log"
	"os"
	"path/filepath"
	"pulsar_study/domain/entity"
	"pulsar_study/domain/event"
	"pulsar_study/infra/dal"
	"strings"
	"sync"
	"time"

	"github.com/google/wire"
	"github.com/jszwec/csvutil"
	"github.com/oklog/ulid/v2"
	"github.com/tidwall/gjson"
)

var JsonlCmdSet = wire.NewSet(NewJsonlCmdSrv)

type JsonlCmdSrv struct {
	Publisher *dal.EventPublisher
	fileLocks map[string]*sync.Mutex
	mapMutex  sync.Mutex
}

func NewJsonlCmdSrv(publisher *dal.EventPublisher) *JsonlCmdSrv {
	ins := &JsonlCmdSrv{
		Publisher: publisher,
		fileLocks: make(map[string]*sync.Mutex),
	}
	ins.subscribe()
	return ins
}

func (c *JsonlCmdSrv) SendMessages(ctx context.Context) error {
	file := "../data/user.jsonl"
	reader, err := entity.NewJsonlShardReader(file, 10)
	if err != nil {
		return err
	}
	for {
		records, err := reader.ReadBatch()
		if err != nil {
			return err
		}
		if records == nil || len(records) == 0 {
			break
		}
		for _, record := range records {
			e := event.JsonlLoadedEvent{
				EntityID:   ulid.Make().String(),
				EventID:    ulid.Make().String(),
				EventTime:  time.Now().Local(),
				EventBody:  record.String(),
				EventType:  "send user json",
				EventState: 0,
			}
			//err = c.Publisher.Publish(ctx, e)
			err = c.Publisher.AsyncPublish(ctx, e)
			if err != nil {
				return err
			}
		}
	}
	return nil
}

func (c *JsonlCmdSrv) ProcessMessages(ctx context.Context, msg *entity.EventMessage) error {
	log.Printf("开始处理消息: %s", msg.ID)
	time.Sleep(10 * time.Second) // 改为10秒，更容易测试
	const outputPrefix = "/Users/rdn/Documents/study/study/data/"

	var record entity.Record
	err := json.Unmarshal(
		[]byte(gjson.Get(msg.Content, "event_body").String()),
		&record,
	)
	if err != nil {
		return err
	}

	var file *os.File
	filename := filepath.Base(record.File)
	outputFileName := strings.TrimSuffix(filename, filepath.Ext(filename))
	outputFile := outputPrefix + outputFileName + ".csv"

	fileLock := c.getFileLock(outputFile)
	fileLock.Lock()
	defer fileLock.Unlock()

	if record.Start {
		file, err = os.Create(outputFile)
	} else {
		file, err = os.OpenFile(outputFile, os.O_RDWR|os.O_APPEND, 0644)
	}
	if err != nil {
		return err
	}

	defer file.Close()

	writer := csv.NewWriter(file)
	if record.Start {
		// 对于每个文件的开头文件，先写入 csv 的头部
		header, err := csvutil.Header(entity.User{}, "csv")
		if err != nil {
			return err
		}
		if err = writer.Write(header); err != nil {
			return err
		}
		writer.Flush()
		if err = writer.Error(); err != nil {
			return err
		}
	}

	// 将每个 user 结构体转换为 csv 的一行
	encoder := csvutil.NewEncoder(writer)
	encoder.AutoHeader = false
	if err = encoder.Encode(record.User.ToCsv()); err != nil {
		return err
	}
	writer.Flush()
	if err = writer.Error(); err != nil {
		return err
	}
	if record.Done {
		log.Printf("文件:%s 已经处理完毕!", record.File)
	}
	return nil
}

func (c *JsonlCmdSrv) subscribe() {
	ctx := context.Background()
	c.Publisher.Subscribe(ctx, event.JsonlLoadedEvent{}, c.ProcessMessages)
}

func (c *JsonlCmdSrv) getFileLock(file string) *sync.Mutex {
	c.mapMutex.Lock()
	defer c.mapMutex.Unlock()

	if lock, ok := c.fileLocks[file]; ok {
		return lock
	}

	lock := new(sync.Mutex)
	c.fileLocks[file] = lock
	return lock
}
